package com.afdor.rws.event.driven.support.transaction.impl;

import com.afdor.rws.event.driven.EventSourceManager;
import com.afdor.rws.event.driven.channel.transaction.ChannelTransactionItem;
import com.afdor.rws.event.driven.entities.AggregationEvent;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.support.transaction.EventBusTransactionItem;
import lombok.Getter;
import lombok.NonNull;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 事件总线事物管理器默认实现方式
 *
 * @author 悭梵
 * @date Created in 2018-12-17 16:57
 */
public class DefaultEventBusTransactionItemImpl implements EventBusTransactionItem {
    /**
     * 事物项标识
     */
    @Getter
    private final String transactionId;
    /**
     * 是否已提交成功
     */
    private final AtomicBoolean isCommited;
    /**
     * 是否已执行回滚操作
     */
    private final AtomicBoolean isRollback;

    @Getter
    private final EventSourceManager eventSourceManager;
    /**
     * 子事物管理器信息
     */
    private final Map<String, ChannelTransactionItem> channelTransactionItemMap;
    /**
     * 事物事件队列
     */
    @Getter
    private final AggregationEvent aggregationEvent = new AggregationEvent.EventBuilder().build();


    public DefaultEventBusTransactionItemImpl(@NonNull final String transactionId, final EventSourceManager eventSourceManager) {
        this.transactionId = transactionId;
        this.isCommited = new AtomicBoolean();
        this.isRollback = new AtomicBoolean();
        this.eventSourceManager = eventSourceManager;
        this.channelTransactionItemMap = new HashMap();
    }

    @Override
    public synchronized void addEvent(@NonNull Event event) {
        final String eventSourceName = event.getEventSourceName();
        ChannelTransactionItem channelTransactionItem = this.channelTransactionItemMap.get(eventSourceName);
        if (channelTransactionItem == null) {
            ChannelTransactionItem newChannelTransactionItem = getEventSourceManager()
                    .getEventSource(eventSourceName)
                    .getChannelProvider()
                    .newChannelTransaction();

            this.channelTransactionItemMap.putIfAbsent(eventSourceName, newChannelTransactionItem);
            channelTransactionItem = this.channelTransactionItemMap.get(eventSourceName);
        }
        channelTransactionItem.addEvent(event);
        this.aggregationEvent.addEvent(event);
    }

    @Override
    public void rollback() {
        if (isCommited.get()) {
            throw new IllegalStateException("事务管理器已执行提交操作，不能再执行回滚操作了。");
        }
        if (!isRollback.get()) {
            isRollback.compareAndSet(Boolean.FALSE, Boolean.TRUE);
            this.channelTransactionItemMap.entrySet()
                    .stream()
                    .forEach(entry -> entry.getValue().rollback());
            this.aggregationEvent.rollback();
        }
    }

    @Override
    public boolean commit() {
        if (isRollback.get()) {
            throw new IllegalStateException("事务管理器已执行回滚操作，不能再执行提交操作了。");
        }
        if (!isCommited.get()) {
            isCommited.compareAndSet(Boolean.FALSE, Boolean.TRUE);
            this.aggregationEvent.setSendType(Event.SendType.ASYNC_TRANSACTION, Event.SendType.ASYNC);
            boolean status = this.channelTransactionItemMap.entrySet()
                    .parallelStream()
                    .map(entry -> entry.getValue().commit())
                    .allMatch(Boolean::booleanValue);
            this.aggregationEvent.commit();
            return status;
        }
        return Boolean.TRUE;
    }
}
